// FraudDetectionReducer.java
package com.taxi.analysis.fraud;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 异常行程检测Reducer
 * 功能：Join行程数据和地点统计数据，进行异常检测分析
 * 技术：Join + 条件过滤 + 异常检测算法
 */
public class FraudDetectionReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        // 分离行程数据和地点数据
        List<TripRecord> trips = new ArrayList<>();
        LocationStats locationStats = null;

        // 解析所有输入数据
        for (Text value : values) {
            try {
                String record = value.toString();
                String[] parts = record.split("\\|", 2);

                if (parts.length >= 2) {
                    String dataType = parts[0];
                    String data = parts[1];

                    if ("TRIP".equals(dataType)) {
                        TripRecord trip = parseTripRecord(data);
                        if (trip != null) {
                            trips.add(trip);
                        }
                    } else if ("LOCATION".equals(dataType)) {
                        locationStats = parseLocationStats(data);
                    }
                }

            } catch (Exception e) {
                System.err.println("解析数据错误: " + value.toString());
            }
        }

        // 如果有地点统计数据和行程数据，进行Join分析
        if (locationStats != null && !trips.isEmpty()) {
            performFraudDetection(key.toString(), trips, locationStats, context);
        }
    }

    /**
     * 执行异常检测分析
     */
    private void performFraudDetection(String joinKey, List<TripRecord> trips,
                                       LocationStats locationStats, Context context)
            throws IOException, InterruptedException {

        // 1. 基础统计分析
        FraudAnalysis analysis = performBasicAnalysis(trips, locationStats);

        // 2. 异常模式检测
        detectAnomalyPatterns(analysis, trips, locationStats);

        // 3. 风险评估
        assessRiskLevel(analysis, locationStats);

        // 4. 生成检测报告
        generateDetectionReport(joinKey, analysis, context);

        // 5. 输出可疑行程详情
        outputSuspiciousTrips(joinKey, trips, locationStats, context);
    }

    /**
     * 执行基础统计分析
     */
    private FraudAnalysis performBasicAnalysis(List<TripRecord> trips, LocationStats locationStats) {
        FraudAnalysis analysis = new FraudAnalysis();

        analysis.totalTrips = trips.size();
        analysis.locationId = locationStats.locationId;
        analysis.locationType = locationStats.locationType;
        analysis.locationRiskLevel = locationStats.riskLevel;

        // 统计各类异常
        int speedAnomalies = 0;
        int fareAnomalies = 0;
        int ratioAnomalies = 0;
        int durationAnomalies = 0;
        int suspiciousTrips = 0;

        double totalFare = 0;
        double totalDistance = 0;
        double totalDuration = 0;

        for (TripRecord trip : trips) {
            if (trip.speedAnomaly) speedAnomalies++;
            if (trip.fareAnomaly) fareAnomalies++;
            if (trip.ratioAnomaly) ratioAnomalies++;
            if (trip.durationAnomaly) durationAnomalies++;
            if (trip.isSuspicious) suspiciousTrips++;

            totalFare += trip.totalAmount;
            totalDistance += trip.tripDistance;
            totalDuration += trip.tripDuration;
        }

        analysis.speedAnomalies = speedAnomalies;
        analysis.fareAnomalies = fareAnomalies;
        analysis.ratioAnomalies = ratioAnomalies;
        analysis.durationAnomalies = durationAnomalies;
        analysis.suspiciousTrips = suspiciousTrips;

        analysis.avgFare = totalFare / trips.size();
        analysis.avgDistance = totalDistance / trips.size();
        analysis.avgDuration = totalDuration / trips.size();

        analysis.suspiciousRate = (double) suspiciousTrips / trips.size();

        return analysis;
    }

    /**
     * 检测异常模式
     */
    private void detectAnomalyPatterns(FraudAnalysis analysis, List<TripRecord> trips,
                                       LocationStats locationStats) {

        // 1. 检测时间集中模式（短时间内大量异常行程）
        Map<Integer, Integer> hourlyAnomalies = new HashMap<>();
        for (TripRecord trip : trips) {
            if (trip.isSuspicious) {
                hourlyAnomalies.put(trip.pickupHour,
                        hourlyAnomalies.getOrDefault(trip.pickupHour, 0) + 1);
            }
        }

        // 找出异常集中的时段
        int maxHourlyAnomalies = hourlyAnomalies.values().stream().mapToInt(Integer::intValue).max().orElse(0);
        analysis.maxHourlyAnomalies = maxHourlyAnomalies;
        analysis.hasTimeConcentration = maxHourlyAnomalies >= Math.max(3, analysis.suspiciousTrips * 0.3);

        // 2. 检测费用异常模式
        double avgFareDeviation = Math.abs(analysis.avgFare - locationStats.avgFare);
        double fareDeviationThreshold = locationStats.stdFare * 2;
        analysis.hasFareDeviation = avgFareDeviation > fareDeviationThreshold;

        // 3. 检测距离异常模式
        double avgDistanceDeviation = Math.abs(analysis.avgDistance - locationStats.avgDistance);
        double distanceDeviationThreshold = locationStats.stdDistance * 2;
        analysis.hasDistanceDeviation = avgDistanceDeviation > distanceDeviationThreshold;

        // 4. 检测速度异常模式
        double speedAnomalyRate = (double) analysis.speedAnomalies / analysis.totalTrips;
        analysis.hasSpeedPattern = speedAnomalyRate > 0.2; // 超过20%的行程有速度异常

        // 5. 检测复合异常模式（多种异常同时出现）
        int multipleAnomalyTrips = 0;
        for (TripRecord trip : trips) {
            int anomalyCount = 0;
            if (trip.speedAnomaly) anomalyCount++;
            if (trip.fareAnomaly) anomalyCount++;
            if (trip.ratioAnomaly) anomalyCount++;
            if (trip.durationAnomaly) anomalyCount++;

            if (anomalyCount >= 2) {
                multipleAnomalyTrips++;
            }
        }

        analysis.multipleAnomalyTrips = multipleAnomalyTrips;
        analysis.hasMultipleAnomalyPattern = multipleAnomalyTrips >= Math.max(2, analysis.totalTrips * 0.1);
    }

    /**
     * 评估风险等级
     */
    private void assessRiskLevel(FraudAnalysis analysis, LocationStats locationStats) {
        int riskScore = 0;

        // 基于可疑行程比例评分
        if (analysis.suspiciousRate >= 0.5) {
            riskScore += 4;
        } else if (analysis.suspiciousRate >= 0.3) {
            riskScore += 3;
        } else if (analysis.suspiciousRate >= 0.15) {
            riskScore += 2;
        } else if (analysis.suspiciousRate >= 0.05) {
            riskScore += 1;
        }

        // 基于地点风险等级评分
        switch (locationStats.riskLevel) {
            case "高风险":
                riskScore += 3;
                break;
            case "中风险":
                riskScore += 2;
                break;
            case "低风险":
                riskScore += 1;
                break;
        }

        // 基于异常模式评分
        if (analysis.hasTimeConcentration) riskScore += 2;
        if (analysis.hasFareDeviation) riskScore += 2;
        if (analysis.hasDistanceDeviation) riskScore += 1;
        if (analysis.hasSpeedPattern) riskScore += 2;
        if (analysis.hasMultipleAnomalyPattern) riskScore += 3;

        // 基于异常数量评分
        double totalAnomalyRate = (double) (analysis.speedAnomalies + analysis.fareAnomalies +
                analysis.ratioAnomalies + analysis.durationAnomalies) /
                (analysis.totalTrips * 4);
        if (totalAnomalyRate >= 0.3) {
            riskScore += 2;
        } else if (totalAnomalyRate >= 0.15) {
            riskScore += 1;
        }

        // 确定最终风险等级
        if (riskScore >= 12) {
            analysis.overallRiskLevel = "极高风险";
        } else if (riskScore >= 8) {
            analysis.overallRiskLevel = "高风险";
        } else if (riskScore >= 5) {
            analysis.overallRiskLevel = "中等风险";
        } else if (riskScore >= 2) {
            analysis.overallRiskLevel = "低风险";
        } else {
            analysis.overallRiskLevel = "正常";
        }

        analysis.riskScore = riskScore;
    }

    /**
     * 生成检测报告
     */
    private void generateDetectionReport(String joinKey, FraudAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        // 1. 基础统计报告
        String basicReport = String.format("基础统计|地点:%s|类型:%s|总行程:%d|可疑行程:%d|可疑率:%.3f|平均费用:%.2f|平均距离:%.2f|平均时长:%.2f",
                analysis.locationId, analysis.locationType, analysis.totalTrips,
                analysis.suspiciousTrips, analysis.suspiciousRate, analysis.avgFare,
                analysis.avgDistance, analysis.avgDuration);

        context.write(new Text("基础统计_" + joinKey), new Text(basicReport));

        // 2. 异常类型统计报告
        String anomalyReport = String.format("异常统计|速度异常:%d|费用异常:%d|比例异常:%d|时长异常:%d|复合异常:%d|最大小时异常:%d",
                analysis.speedAnomalies, analysis.fareAnomalies, analysis.ratioAnomalies,
                analysis.durationAnomalies, analysis.multipleAnomalyTrips, analysis.maxHourlyAnomalies);

        context.write(new Text("异常统计_" + joinKey), new Text(anomalyReport));

        // 3. 模式检测报告
        String patternReport = String.format("模式检测|时间集中:%s|费用偏离:%s|距离偏离:%s|速度模式:%s|复合模式:%s",
                analysis.hasTimeConcentration ? "是" : "否",
                analysis.hasFareDeviation ? "是" : "否",
                analysis.hasDistanceDeviation ? "是" : "否",
                analysis.hasSpeedPattern ? "是" : "否",
                analysis.hasMultipleAnomalyPattern ? "是" : "否");

        context.write(new Text("模式检测_" + joinKey), new Text(patternReport));

        // 4. 风险评估报告
        String riskReport = String.format("风险评估|地点风险:%s|整体风险:%s|风险评分:%d|建议措施:%s",
                analysis.locationRiskLevel, analysis.overallRiskLevel, analysis.riskScore,
                generateRecommendation(analysis));

        context.write(new Text("风险评估_" + joinKey), new Text(riskReport));
    }

    /**
     * 输出可疑行程详情
     */
    private void outputSuspiciousTrips(String joinKey, List<TripRecord> trips,
                                       LocationStats locationStats, Context context)
            throws IOException, InterruptedException {

        int outputCount = 0;
        for (TripRecord trip : trips) {
            if (trip.isSuspicious && outputCount < 10) { // 只输出前10个可疑行程
                String suspiciousDetail = String.format("可疑行程|行程ID:%s|距离:%.2f|费用:%.2f|时长:%.2f|速度:%.2f|单价:%.2f|异常类型:%s",
                        trip.tripId, trip.tripDistance, trip.totalAmount,
                        trip.tripDuration, trip.avgSpeed, trip.farePerMile,
                        getAnomalyTypes(trip));

                context.write(new Text("可疑行程_" + joinKey + "_" + (outputCount + 1)), new Text(suspiciousDetail));
                outputCount++;
            }
        }
    }

    /**
     * 获取异常类型描述
     */
    private String getAnomalyTypes(TripRecord trip) {
        List<String> anomalies = new ArrayList<>();
        if (trip.speedAnomaly) anomalies.add("速度");
        if (trip.fareAnomaly) anomalies.add("费用");
        if (trip.ratioAnomaly) anomalies.add("比例");
        if (trip.durationAnomaly) anomalies.add("时长");

        return anomalies.isEmpty() ? "其他" : String.join(",", anomalies);
    }

    /**
     * 生成建议措施
     */
    private String generateRecommendation(FraudAnalysis analysis) {
        if (analysis.riskScore >= 12) {
            return "立即调查,暂停服务";
        } else if (analysis.riskScore >= 8) {
            return "重点监控,人工审核";
        } else if (analysis.riskScore >= 5) {
            return "加强监控,定期检查";
        } else if (analysis.riskScore >= 2) {
            return "常规监控,关注变化";
        } else {
            return "正常监控";
        }
    }

    /**
     * 解析行程记录
     */
    private TripRecord parseTripRecord(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 16) {
                TripRecord record = new TripRecord();

                record.tripId = fields[0];
                record.joinType = fields[1];
                record.pickupLocationId = fields[2];
                record.dropoffLocationId = fields[3];
                record.tripDistance = Double.parseDouble(fields[4]);
                record.totalAmount = Double.parseDouble(fields[5]);
                record.passengerCount = Integer.parseInt(fields[6]);
                record.tripDuration = Double.parseDouble(fields[7]);
                record.avgSpeed = Double.parseDouble(fields[8]);
                record.farePerMile = Double.parseDouble(fields[9]);
                record.pickupHour = Integer.parseInt(fields[10]);
                record.speedAnomaly = "1".equals(fields[11]);
                record.fareAnomaly = "1".equals(fields[12]);
                record.ratioAnomaly = "1".equals(fields[13]);
                record.durationAnomaly = "1".equals(fields[14]);
                record.isSuspicious = "1".equals(fields[15]);

                return record;
            }
        } catch (Exception e) {
            System.err.println("解析行程记录错误: " + data);
        }
        return null;
    }

    /**
     * 解析地点统计数据
     */
    private LocationStats parseLocationStats(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 14) {
                LocationStats stats = new LocationStats();

                stats.locationId = fields[0];
                stats.totalTrips = Integer.parseInt(fields[1]);
                stats.suspiciousTrips = Integer.parseInt(fields[2]);
                stats.suspiciousRate = Double.parseDouble(fields[3]);
                stats.avgFare = Double.parseDouble(fields[4]);
                stats.stdFare = Double.parseDouble(fields[5]);
                stats.avgDistance = Double.parseDouble(fields[6]);
                stats.stdDistance = Double.parseDouble(fields[7]);
                stats.avgSpeedStat = Double.parseDouble(fields[8]);
                stats.stdSpeed = Double.parseDouble(fields[9]);
                stats.avgFarePerMile = Double.parseDouble(fields[10]);
                stats.stdFarePerMile = Double.parseDouble(fields[11]);
                stats.locationType = fields[12];
                stats.riskLevel = fields[13];

                return stats;
            }
        } catch (Exception e) {
            System.err.println("解析地点统计错误: " + data);
        }
        return null;
    }

    /**
     * 行程记录内部类
     */
    private static class TripRecord {
        String tripId;
        String joinType;
        String pickupLocationId;
        String dropoffLocationId;
        double tripDistance;
        double totalAmount;
        int passengerCount;
        double tripDuration;
        double avgSpeed;
        double farePerMile;
        int pickupHour;
        boolean speedAnomaly;
        boolean fareAnomaly;
        boolean ratioAnomaly;
        boolean durationAnomaly;
        boolean isSuspicious;
    }

    /**
     * 地点统计内部类
     */
    private static class LocationStats {
        String locationId;
        int totalTrips;
        int suspiciousTrips;
        double suspiciousRate;
        double avgFare;
        double stdFare;
        double avgDistance;
        double stdDistance;
        double avgSpeedStat;
        double stdSpeed;
        double avgFarePerMile;
        double stdFarePerMile;
        String locationType;
        String riskLevel;
    }

    /**
     * 异常分析结果内部类
     */
    private static class FraudAnalysis {
        String locationId;
        String locationType;
        String locationRiskLevel;
        int totalTrips;
        int suspiciousTrips;
        double suspiciousRate;
        int speedAnomalies;
        int fareAnomalies;
        int ratioAnomalies;
        int durationAnomalies;
        int multipleAnomalyTrips;
        int maxHourlyAnomalies;
        double avgFare;
        double avgDistance;
        double avgDuration;
        boolean hasTimeConcentration;
        boolean hasFareDeviation;
        boolean hasDistanceDeviation;
        boolean hasSpeedPattern;
        boolean hasMultipleAnomalyPattern;
        String overallRiskLevel;
        int riskScore;
    }
}
